package com.nian.message.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.util.List;

/**
 * 用途：
 * 开发：zhangmj
 * 日期：2020/4/10 0010 15:51
 */
public class TestConsumer {

    @Test
    public void consumer1() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("nian-message-service-consumer-group");
        consumer.setNamesrvAddr("192.168.203.132:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        try {
            String t = "producer_topic_shiwu_test_";
            String y = "producer_topic_test";
            consumer.subscribe(t, "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    try {
                        for (MessageExt msg:msgs) {
                            System.out.println("---------------------------------------------");
                            System.out.println("事务ID：" + msg.getMsgId());
                            System.out.println("事务ID：" + msg.getTransactionId());
                            System.out.println("transactionInfo：" + new String(msg.getBody()));
                        }

                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
                }
            });

            consumer.start();
            System.out.println("transaction_consumer started.");
            while (true) {
            }
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
